/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.msgobject;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.global.GlobalParas;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;

/**
 *
 * @author YG
 */
public class Message implements Runnable {

    public Logger logger = Logger.getLogger(Message.class.getName());
    public String[] colNameList;
    public REFieldType[] colTypeList;
    public List<Object[]> data;
//    public String docSchemaName;
    public Map<String, String> userDesc;

    @Override
    public void run() {

        while (true) {
            try {
//                List<REMessage> msglist = new ArrayList<>();
//                while (true) {
                REMessage msg = GlobalParas.oriMsgPool.poll(3, TimeUnit.SECONDS);
                if (msg == null) {
                    if (GlobalParas.killRemsg2Msg) {
                        break;
                    }
                    continue;
                }
                int count = 0;
                int flag = 0;
                colNameList = msg.colNames;
                colTypeList = msg.types;
                data = msg.data;
                Random random = new Random();
                for (Object[] line : data) {
                    Map<String, LinkedList> map = new HashMap<>();
                    for (int i = 0; i < colNameList.length; i++) {
                        List list = new LinkedList();
                        list.add(colTypeList[i]);
                        list.add(line[i]);
                        map.put(colNameList[i], (LinkedList) list);
                    }
                    if (random.nextInt(100) < GlobalParas.divideNum) {
                        boolean offer = GlobalParas.dwdMsgPool.offer((HashMap) map);
                        while (!offer) {
                            Thread.sleep(500);
                            offer = GlobalParas.dwdMsgPool.offer((HashMap) map);
                        }
                    } else {
                        boolean offer = GlobalParas.sendMsgPool2.offer((HashMap) map);
                        while (!offer) {
                            Thread.sleep(500);
                            offer = GlobalParas.sendMsgPool2.offer((HashMap) map);
                        }
                    }
                }

                if (GlobalParas.killRemsg2Msg) {
                    break;
                }
            } catch (InterruptedException ex) {
                logger.error(ex, ex);
            }
        }
        if (!GlobalParas.ReMsg2MsgQueue.isEmpty()) {
            GlobalParas.ReMsg2MsgQueue.poll();
            logger.info("GlobalParas.ReMsg2MsgQueue size is : " + GlobalParas.ReMsg2MsgQueue.size());
        }

        logger.info(
                "ReMsg 2 Msg exit ");
    }
}
